package com.reactim.message.connection;

import com.reactim.message.config.ConnectionConfig;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.time.Duration;
import java.time.Instant;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 故障恢复服务
 * 负责Redis连接的健康检查、故障检测和自动恢复
 */
@Service
@Slf4j
@RequiredArgsConstructor
public class FaultRecoveryService {

    private final ReactiveStringRedisTemplate redisTemplate;
    private final ConnectionConfig connectionConfig;

    // 故障状态跟踪
    private volatile boolean redisHealthy = true;
    private final AtomicInteger consecutiveFailures = new AtomicInteger(0);
    private volatile Instant lastHealthCheckTime = Instant.now();
    private volatile Instant lastFailureTime;
    private volatile Instant lastRecoveryTime;

    // 统计信息
    private final AtomicLong totalFailures = new AtomicLong(0);
    private final AtomicLong totalRecoveries = new AtomicLong(0);
    private final AtomicLong totalFailureDurationSeconds = new AtomicLong(0);

    // 故障转移存储（本地内存）
    private final ConcurrentHashMap<Long, ConnectionDetails> failoverConnectionStorage = new ConcurrentHashMap<>();

    /**
     * 定期健康检查
     */
    @Scheduled(fixedDelayString = "#{connectionConfig.faultRecovery.healthCheckIntervalMs}")
    public void performHealthCheck() {
        healthCheck().subscribe(
            healthy -> {
                if (healthy && !redisHealthy) {
                    // Redis恢复健康
                    handleRedisRecovery();
                } else if (!healthy && redisHealthy) {
                    // Redis出现故障
                    handleRedisFailure();
                }
            },
            error -> {
                log.error("健康检查执行失败", error);
                handleRedisFailure();
            }
        );
    }

    /**
     * 执行健康检查
     */
    private Mono<Boolean> healthCheck() {
        lastHealthCheckTime = Instant.now();

        return redisTemplate.opsForValue()
                .set("health:check", "ping")
                .timeout(Duration.ofMillis(connectionConfig.getFaultRecovery().getRecoveryTimeoutMs()))
                .map(result -> {
//                    log.debug("Redis健康检查成功");
                    return true;
                })
                .onErrorReturn(false);
    }

    /**
     * 处理Redis故障
     */
    private void handleRedisFailure() {
        int failures = consecutiveFailures.incrementAndGet();
        lastFailureTime = Instant.now();

        log.warn("Redis健康检查失败，连续失败次数: {}", failures);

        if (failures >= connectionConfig.getFaultRecovery().getFailureThreshold()) {
            if (redisHealthy) {
                redisHealthy = false;
                totalFailures.incrementAndGet();
                log.error("Redis被标记为不健康，启用故障转移模式");
            }
        }
    }

    /**
     * 处理Redis恢复
     */
    private void handleRedisRecovery() {
        redisHealthy = true;
        consecutiveFailures.set(0);
        lastRecoveryTime = Instant.now();
        totalRecoveries.incrementAndGet();

        if (lastFailureTime != null) {
            long failureDuration = Duration.between(lastFailureTime, lastRecoveryTime).getSeconds();
            totalFailureDurationSeconds.addAndGet(failureDuration);
        }

        log.info("Redis恢复健康，开始数据同步");

        // 异步执行数据恢复
        recoverDataFromFailover().subscribe(
            success -> log.info("故障转移数据恢复完成: {}", success),
            error -> log.error("故障转移数据恢复失败", error)
        );
    }

    /**
     * 从故障转移存储恢复数据到Redis
     */
    private Mono<Boolean> recoverDataFromFailover() {
        if (failoverConnectionStorage.isEmpty()) {
            return Mono.just(true);
        }

        return Mono.fromCallable(() -> {
            int recoveredCount = 0;
            for (Map.Entry<Long, ConnectionDetails> entry : failoverConnectionStorage.entrySet()) {
                try {
                    // 这里应该调用实际的连接管理器来恢复数据
                    // 为了简化，我们只是记录日志
                    log.debug("恢复连接数据: userId={}", entry.getKey());
                    recoveredCount++;
                } catch (Exception e) {
                    log.error("恢复连接数据失败: userId={}", entry.getKey(), e);
                }
            }

            // 清空故障转移存储
            failoverConnectionStorage.clear();

            log.info("数据恢复完成，恢复了 {} 个连接", recoveredCount);
            return true;
        });
    }

    /**
     * 在故障转移存储中存储连接
     */
    public Mono<Boolean> storeConnectionInFailover(Long userId, ConnectionDetails connectionDetails) {
        return Mono.fromCallable(() -> {
            failoverConnectionStorage.put(userId, connectionDetails);
            log.debug("连接已存储到故障转移存储: userId={}", userId);
            return true;
        });
    }

    /**
     * 从故障转移存储获取连接
     */
    public Mono<ConnectionDetails> getConnectionFromFailover(Long userId) {
        return Mono.fromCallable(() -> {
            ConnectionDetails details = failoverConnectionStorage.get(userId);
            log.debug("从故障转移存储获取连接: userId={}, found={}", userId, details != null);
            return details;
        });
    }

    /**
     * 从故障转移存储移除连接
     */
    public Mono<Boolean> removeConnectionFromFailover(Long userId) {
        return Mono.fromCallable(() -> {
            ConnectionDetails removed = failoverConnectionStorage.remove(userId);
            log.debug("从故障转移存储移除连接: userId={}, removed={}", userId, removed != null);
            return removed != null;
        });
    }

    /**
     * 手动触发健康检查
     */
    public Mono<Boolean> manualHealthCheck() {
        return healthCheck()
                .doOnNext(healthy -> {
                    if (healthy && !redisHealthy) {
                        handleRedisRecovery();
                    } else if (!healthy && redisHealthy) {
                        handleRedisFailure();
                    }
                });
    }

    /**
     * 手动触发数据恢复
     */
    public Mono<Boolean> manualRecovery() {
        if (!redisHealthy) {
            return Mono.just(false);
        }
        return recoverDataFromFailover();
    }

    /**
     * 获取Redis健康状态
     */
    public boolean isRedisHealthy() {
        return redisHealthy;
    }

    /**
     * 获取连续失败次数
     */
    public int getConsecutiveFailures() {
        return consecutiveFailures.get();
    }

    /**
     * 获取故障恢复统计信息
     */
    public FaultRecoveryStats getStats() {
        long averageFailureDuration = 0;
        if (totalFailures.get() > 0) {
            averageFailureDuration = totalFailureDurationSeconds.get() / totalFailures.get();
        }

        return FaultRecoveryStats.builder()
                .redisHealthy(redisHealthy)
                .consecutiveFailures(consecutiveFailures.get())
                .lastHealthCheckTime(lastHealthCheckTime)
                .lastFailureTime(lastFailureTime)
                .lastRecoveryTime(lastRecoveryTime)
                .failoverConnectionCount(failoverConnectionStorage.size())
                .failoverOnlineUserCount(failoverConnectionStorage.size()) // 简化实现
                .totalFailures(totalFailures.get())
                .totalRecoveries(totalRecoveries.get())
                .averageFailureDurationSeconds(averageFailureDuration)
                .build();
    }

    /**
     * 重置统计信息
     */
    public void resetStats() {
        totalFailures.set(0);
        totalRecoveries.set(0);
        totalFailureDurationSeconds.set(0);
        consecutiveFailures.set(0);
        redisHealthy = true;
        lastFailureTime = null;
        lastRecoveryTime = null;
        log.info("故障恢复统计信息已重置");
    }

    /**
     * 获取故障转移存储中的所有连接用户ID
     */
    public Mono<java.util.Set<Long>> getFailoverConnections() {
        return Mono.fromCallable(() -> {
            java.util.Set<Long> userIds = new java.util.HashSet<>(failoverConnectionStorage.keySet());
            log.debug("获取故障转移连接用户ID: count={}", userIds.size());
            return userIds;
        });
    }
}
